package com.jacky.grpc.serices;

import com.google.common.collect.Lists;
import com.jacky.grpc.proto.HelloReply;
import com.jacky.grpc.proto.HelloRequest;
import com.jacky.grpc.proto.SimpleGrpc;
import io.grpc.stub.StreamObserver;
import net.devh.boot.grpc.server.service.GrpcService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.stream.IntStream;

@GrpcService
public class HelloGrpcServiceImpl extends SimpleGrpc.SimpleImplBase {

    private static final Logger LOGGER = LoggerFactory.getLogger(HelloGrpcServiceImpl.class);


    @Override
    public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
        LOGGER.info("[sayHello] receive: {}", request);
        HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + request.getName()).build();
        responseObserver.onNext(reply);
        LOGGER.info("[sayHello] reply: {}", reply);
        responseObserver.onCompleted();
    }

    @Override
    public void sayHelloWithServerStream(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
        LOGGER.info("[sayHelloWithServerStream] receive: {}", request);
        // 服务端输出响应流(多次onNext输出响应结果)
        IntStream.range(0, 5).forEach(num -> {
            HelloReply reply = HelloReply.newBuilder().setMessage(String.format("greeting loop_%d for %s", num, request.getName())).build();
            LOGGER.info("[sayHelloWithServerStream] reply:{}", reply);
            responseObserver.onNext(reply);
        });

        responseObserver.onCompleted();
    }

    @Override
    public StreamObserver<HelloRequest> sayHelloWithClientStream(StreamObserver<HelloReply> responseObserver) {
        // 实现StreamObserver接受客户端流
        return new StreamObserver<>() {
            private final List<String> nameList = Lists.newArrayList();

            @Override
            public void onNext(HelloRequest req) {
                nameList.add(req.getName());
                LOGGER.info("[sayHelloWithClientStream] has receive:{}, now receive:{}", nameList.size(), req.getName());
            }

            @Override
            public void onError(Throwable t) {
                LOGGER.error("[sayHelloWithClientStream] cause error:", t);
            }

            @Override
            public void onCompleted() {
                // 构造返回结果
                String nameStr = String.join(",", nameList);
                HelloReply reply = HelloReply.newBuilder().setMessage(String.format("Hello %s", nameStr)).build();
                LOGGER.info("[sayHelloWithClientStream] reply:{}", reply);
                responseObserver.onNext(reply);
                responseObserver.onCompleted();
            }
        };
    }

    @Override
    public StreamObserver<HelloRequest> sayHelloBiDirectStream(StreamObserver<HelloReply> responseObserver) {
        return new StreamObserver<HelloRequest>() {
            @Override
            public void onNext(HelloRequest req) {
                LOGGER.info("[sayHelloBiDirectStream] receive: {}", req.getName());
                HelloReply build = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
                responseObserver.onNext(build);
            }

            @Override
            public void onError(Throwable t) {
                LOGGER.error("[sayHelloBiDirectStream] cause error:", t);
            }

            @Override
            public void onCompleted() {
                LOGGER.info("[sayHelloBiDirectStream] completed");
                responseObserver.onCompleted();
            }
        };
    }
}
